package cn.com.server.handler;

import cn.com.common.Invocation;
import cn.com.common.Request;
import cn.com.common.model.ProbeModel;
import cn.com.rpc.channel.Channel;
import cn.com.rpc.handler.Handler;
import cn.com.rule.RuleDictionary;
import cn.com.common.RuleModel;
import cn.com.server.cache.WindowCacheHolder;
import cn.com.server.metric.ArrayMetric;
import cn.com.server.metric.BucketSlideArray;
import cn.com.server.push.Push;
import com.github.benmanes.caffeine.cache.Cache;
import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger;

import java.util.List;

/**
 * @author jiaming
 */
public class ZhuRongHandler implements Handler {

    protected static final Logger log = LogManager.getLogger();

    private Push push;

    public ZhuRongHandler(Push push) {
        this.push = push;
    }

    @Override
    public void connect(Channel channel, Object msg) {
    }

    @Override
    public void disconnect(Channel channel, Object msg) {
    }

    @Override
    public void received(Channel channel, Object msg) {
        //处理请求
        if (msg instanceof Request) {
            Request request = (Request) msg;
            Invocation invocation = (Invocation) request.getData();
            List<ProbeModel> probeModels = (List<ProbeModel>) invocation.getArgs()[0];
            if (probeModels != null && !probeModels.isEmpty()) {
                for (ProbeModel probeModel : probeModels) {
                    doReceived(probeModel);
                }
            }
        }
    }

    /**
     * 接收并推送数据
     *
     * @param probeModel
     */
    private void doReceived(ProbeModel probeModel) {
        //获取规则
        RuleModel rule = RuleDictionary.findRule(probeModel.getKey());
        if (rule == null) {
            log.error("rule is null,key:{}", probeModel.getKey());
            return;
        }
        String key = buildCacheKey(probeModel);
        //根据appname获取，如果为空则新建滑动窗口
        Cache<String, ArrayMetric> cache = WindowCacheHolder.getCache(probeModel.getAppName());
        ArrayMetric arrayMetric = cache.getIfPresent(key);
        if (arrayMetric == null) {
            BucketSlideArray bucketSlideArray = new BucketSlideArray(10, rule.getInterval() * 1000);
            arrayMetric = new ArrayMetric(bucketSlideArray);
            cache.put(key, arrayMetric);
        }
        arrayMetric.add(probeModel.getKeySum());
        long sum = arrayMetric.sum();
        long threshold = rule.getThreshold();
        //探测是否大于阈值
        if (sum >= threshold) {
            //大于阈值，删除key
            cache.invalidate(key);
            try {
                //给客户端推送探测出来的key
                push.push(probeModel);
            } catch (Exception e) {
                //推送报错
                log.info("推送报错，key:{}",key);
                log.error(e.getMessage(), e);
            }
        } else {
            //重新放入缓存，刷新缓存过期时间
            cache.put(key, arrayMetric);
            log.info("重新刷新缓存,key:{}|sum:{}|threshold:{}", key,sum,threshold);
        }
    }

    private String buildCacheKey(ProbeModel probeModel) {
        StringBuilder sb = new StringBuilder();
        sb.append(probeModel.getKeyType());
        sb.append("_");
        sb.append(probeModel.getKey());
        return sb.toString();
    }

    @Override
    public void sent(Channel channel, Object result) {

    }

    @Override
    public void cause(Channel channel, Throwable cause) {

    }
}
